package com.alex.transcations;

import java.util.Properties;
import java.util.UUID;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 * @ClassName KafkaProducerInterceptor @Description TODO @Author zjg @Date 2021/12/615:28 @Version
 * 2.0
 */
public class KafkaProducerTrancations {
  public static void main(String[] args) {
    KafkaProducer<String, String> producer = buildKafkaProducer();
    // 初始化事务
    producer.initTransactions();
    // 开启事务
    producer.beginTransaction();
    try {
      for (int i = 0; i < 10; i++) {
        if (i == 8) {
          int j = 10 / 0;
        }
        ProducerRecord<String, String> record =
            new ProducerRecord<String, String>(
                "topic01", "transcation " + i, "transcation value" + i);
        producer.send(record);
        producer.flush();
      }
      // 提交事务
      producer.commitTransaction();
    } catch (Exception e) {
      System.out.println("出错了~" + e.getMessage());
      producer.abortTransaction();
    } finally {
      producer.close();
    }
  }

  public static KafkaProducer<String, String> buildKafkaProducer() {
    //
    // 创建KakfaProducer
    Properties props = new Properties();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "49.233.172.9:9092");
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

    // 必须配置事务ID，必须唯一的
    props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "www" + UUID.randomUUID().toString());
    // 配置Kafka批处理大小
    props.put(ProducerConfig.BATCH_SIZE_CONFIG, 1024);
    // 等待5ms，如果batch中的数据不足1024
    props.put(ProducerConfig.LINGER_MS_CONFIG, 5);
    // 配置kafka的重试机制和acks
    props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
    props.put(ProducerConfig.ACKS_CONFIG, "all");
    props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 20000);

    return new KafkaProducer<String, String>(props);
  }
}
